package com.chengqs.rapid.core.netty.processor;

import com.chengqs.rapid.common.concurrent.queue.flusher.ParallelFlusher;
import com.chengqs.rapid.common.enums.ResponseCode;
import com.chengqs.rapid.core.RapidConfig;
import com.chengqs.rapid.core.context.HttpRequestWrapper;
import com.chengqs.rapid.core.helper.ResponseHelper;
import com.lmax.disruptor.dsl.ProducerType;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http.*;
import lombok.extern.slf4j.Slf4j;

/**
 * <B>主类名称：</B>NettyBatchEventProcessor<BR>
 * <B>概要说明：</B>flusher缓冲队列的核心实现, 最终调用的方法还是要回归到NettyCoreProcessor<BR>
 *
 * @author: chengqs
 * @create: 2025-03-11 14:21:39
 **/
@Slf4j
public class NettyBatchEventProcessor implements NettyProcessor {

    private static final String THREAD_NAME_PREFIX = "rapid-flusher-";

    private RapidConfig rapidConfig;

    private NettyCoreProcessor nettyCoreProcessor;

    private ParallelFlusher<HttpRequestWrapper> parallelFlusher;

    public NettyBatchEventProcessor(RapidConfig rapidConfig, NettyCoreProcessor nettyCoreProcessor) {
        this.rapidConfig = rapidConfig;
        this.nettyCoreProcessor = nettyCoreProcessor;
        ParallelFlusher.Builder<HttpRequestWrapper> builder = new ParallelFlusher.Builder<HttpRequestWrapper>()
                .bufferSize(rapidConfig.getBufferSize())
                .threads(rapidConfig.getProcessThread())
                .producerType(ProducerType.MULTI)
                .namePrefix(THREAD_NAME_PREFIX)
                .waitStrategy(rapidConfig.getATrueWaitStrategy());
        BatchEventProcessorListener batchEventProcessorListener = new BatchEventProcessorListener();
        builder.eventListener(batchEventProcessorListener);
        this.parallelFlusher = builder.build();
    }

    @Override
    public void process(HttpRequestWrapper httpRequestWrapper) {
        this.parallelFlusher.add(httpRequestWrapper);
    }

    @Override
    public void start() {
        this.nettyCoreProcessor.start();
        this.parallelFlusher.start();
    }

    @Override
    public void shutdown() {
        this.nettyCoreProcessor.shutdown();
        this.parallelFlusher.shutdown();
    }

    /**
     * <B>主类名称：</B>BatchEventProcessorListener<BR>
     * <B>概要说明：</B>监听事件的处理核心逻辑<BR>
     * @author chengqs
     * @create 2025-03-13 02:07:11
     */
    public class BatchEventProcessorListener implements ParallelFlusher.EventListener<HttpRequestWrapper> {
        @Override
        public void onEvent(HttpRequestWrapper event) throws Exception {
            nettyCoreProcessor.process(event);
        }

        @Override
        public void onException(Throwable ex, long sequence, HttpRequestWrapper event) {
            HttpRequest request = event.getFullHttpRequest();
            ChannelHandlerContext ctx = event.getCtx();
            try {
                log.error("NettyBatchEventProcessor$BatchEventProcessorListener#onException() 请求处理失败, request: {}, errorMessage: {}",
                        request, ex.getMessage(), ex);
                // 首先构建响应对象
                FullHttpResponse fullHttpResponse = ResponseHelper.getHttpResponse(ResponseCode.INTERNAL_ERROR);
                // 判断是否保持连接
                if (!HttpUtil.isKeepAlive(request)) {
                    ctx.writeAndFlush(fullHttpResponse).addListener(ChannelFutureListener.CLOSE);
                } else {
                    // 如果保持连接, 则需要设置一下响应头：key: CONNECTION,  value: KEEP_ALIVE
                    fullHttpResponse.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
                    ctx.writeAndFlush(fullHttpResponse);
                }
            } catch (Exception e) {
                log.error("NettyBatchEventProcessor$BatchEventProcessorListener#onException() 请求回写失败, request: {}, errorMessage: {}",
                        request, e.getMessage(), e);
            }
        }
    }

    public RapidConfig getRapidConfig() {
        return rapidConfig;
    }
}
